package com.example.framework.mq.kafka.core.service;

import cn.hutool.core.util.ObjectUtil;
import com.example.framework.common.util.GenericUtils;
import com.example.framework.common.util.json.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.List;

@Slf4j
public abstract class KafkaConsumerService<T> {

    public abstract void consume(String topic, T msgBody);

    public void consume(List<ConsumerRecord<?, String>> records) {
        if (ObjectUtil.isEmpty(records)) {
            return;
        }
        String topic = records.getFirst().topic();
        log.info("Kafka收到消息，Topic={}, records={}", topic, records);
        for (ConsumerRecord<?, ?> consumerRecord : records) {
            try {
                consume(topic, JsonUtils.parseObject((String) consumerRecord.value(), getType()));
            } catch (Exception e) {
                log.error("消费消息失败，Topic={}, msg={}, e={}", topic, consumerRecord, e.getMessage());
                throw e;
            }
        }
    }

    public Class<T> getType() {
        return GenericUtils.getGenericClass(getClass(), 0);
    }
}